package com.tcs.kafka.producer.sender;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.util.concurrent.FailureCallback;
import org.springframework.util.concurrent.ListenableFuture;
import org.springframework.util.concurrent.SuccessCallback;

import com.tcs.kafka.producer.sender.balance.ITopicPartition;

public class KafkaSender<K,T> implements IKafkaSender<K, T> {
	
	private final Logger logger = LoggerFactory.getLogger(KafkaSender.class);
	
	private KafkaTemplate<K,T> kafkaTemplate;
	
	private ITopicPartition topicPartition;
	
	@Override
	public void sendMessage(String topic, Integer partition, Long timestamp, K key, T data) {
		try {
			if (kafkaTemplate.getDefaultTopic().equals(topic)) {
			    kafkaTemplate.sendDefault(partition,timestamp,key,data);
			}else {
			    kafkaTemplate.send(topic, partition, timestamp, key, data);
			}
		} catch (Exception e) {
			logger.error("" , e);
		}
	}
	
	@Override
	public void sendMessage(String topic, Integer partition, Long timestamp, K key, T data , SuccessCallback<SendResult<K, T>> successCallback , FailureCallback failureCallback) {
		ListenableFuture<SendResult<K, T>> listenableFuture = null;
        try {
			if (kafkaTemplate.getDefaultTopic().equals(topic)) {
			    listenableFuture =  kafkaTemplate.sendDefault(partition,timestamp,key,data);
			}else {
			    listenableFuture = kafkaTemplate.send(topic, partition, timestamp, key, data);
			}
		} catch (Exception e) {
			logger.error("" , e);
		}
        listenableFuture.addCallback(successCallback, failureCallback);
	}

	@Override
	public void sendMessage(String topic, Integer partition, Long timestamp, K key, T data,
			SuccessCallback<SendResult<K, T>> successCallback, FailureCallback failureCallback, SenderBalance balance) {
		ListenableFuture<SendResult<K, T>> listenableFuture = null;
        try {
        	if (kafkaTemplate.getDefaultTopic().equals(topic)) {
			    listenableFuture =  kafkaTemplate.sendDefault(partition,timestamp,key,data);
			}else {
				partition = topicPartition.getParititon(topic, balance);
			    listenableFuture = kafkaTemplate.send(topic, partition, timestamp, key, data);
			}
		} catch (Exception e) {
			logger.error("topic : " + topic +  "partition : " + partition , e);
		}
        listenableFuture.addCallback(successCallback, failureCallback);
	}

	public void setKafkaTemplate(KafkaTemplate<K, T> kafkaTemplate) {
		this.kafkaTemplate = kafkaTemplate;
	}

	public void setTopicPartition(ITopicPartition topicPartition) {
		this.topicPartition = topicPartition;
	}
}
